package impl

import akka.{Done, NotUsed}
import akka.cluster.sharding.typed.scaladsl.ClusterSharding
import akka.stream.scaladsl.Source
import akka.util.Timeout
import api.{GreetingMessage, HelloWorldService}
import com.lightbend.lagom.scaladsl.api.ServiceCall
import com.lightbend.lagom.scaladsl.api.broker.Topic
import com.lightbend.lagom.scaladsl.api.transport.BadRequest
import com.lightbend.lagom.scaladsl.broker.TopicProducer
import com.lightbend.lagom.scaladsl.persistence.{EventStreamElement, PersistentEntityRegistry}

import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.duration._


/**
 * Created by yz on 17/4/2020
 */
class HelloWorldServiceImpl(clusterSharding: ClusterSharding, persistenEntityRegistry: PersistentEntityRegistry)(
  implicit ec: ExecutionContext) extends HelloWorldService {

  implicit val timeout = Timeout(5.seconds)

  private def entityRef(id: String) = clusterSharding.entityRefFor(HelloWorldState.typeKey, id)

  override def hello(id: String): ServiceCall[NotUsed, String] = ServiceCall {
    _ =>
      Future.successful(
        s"Hello $id!")
    //      val ref = entityRef(id)
    //      ref.ask[Greeting](replyTo => Hello(id, replyTo)).map(greeting => greeting.message)
  }

  override def useGreeting(id: String): ServiceCall[GreetingMessage, Done] = ServiceCall { request =>
    val ref = entityRef(id)
    ref.ask[Confirmation](replyTo => UseGreetingMessage(request.message, replyTo)).map {
      case Accepted => Done
      case _ => throw BadRequest("Can't upgrade the greeting message.")
    }
  }

  override def greetingsTopic(): Topic[api.GreetingMessageChanged] = TopicProducer.singleStreamWithOffset { fromOffset =>
    persistenEntityRegistry.eventStream(HelloWorldEvent.Tag, fromOffset).map(ev => (convertEvent(ev), ev.offset))

  }

  private def convertEvent(helloEvent: EventStreamElement[HelloWorldEvent]) = {
    helloEvent.event match {
      case GreetingMessageChanged(msg) =>
        api.GreetingMessageChanged(helloEvent.entityId, msg)
    }

  }

  override def tick(intervalMs: Int): ServiceCall[String, Source[String, NotUsed]] = ServiceCall { tickMessage =>
    val t = Source.tick(intervalMs.milliseconds,
      intervalMs.milliseconds,
      tickMessage
    ).mapMaterializedValue(_ => NotUsed)
    Future.successful(t)
  }
}
